package cn.com.daimajiangxin.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
// Use java.time.Duration instead of deprecated Time factory methods

import java.time.Duration;

/**
 * 事件驱动示例：使用 Kafka Source + Watermark + KeyedProcessFunction 实现“伪窗口”统计每小时小费总额。
 *
 * Kafka Topic（示例）消息格式：CSV
 * driverId,eventTimeMillis,tip
 * 例如：42,1710003600000,3.5
 */
public class EventDriven {
        public static void main(String[] args) throws Exception {
                StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                // 开启 checkpoint，便于容错（演示值，可按需调整）
                env.enableCheckpointing(10_000);

                // Kafka Source（新连接器 API）
                KafkaSource<TaxiFare> source = KafkaSource.<TaxiFare>builder()
                                .setBootstrapServers("127.0.0.1:9092")
                                .setTopics("fares")
                                .setGroupId("flink-fare-group")
                                .setValueOnlyDeserializer(new TaxiFareDeserializer())
                                .setProperty("enable.auto.commit", "false")  // 手动控制提交
                                .setProperty("session.timeout.ms", "45000")   // 增加会话超时时
                                .build();

                // 绑定事件时间与乱序水位线
                DataStream<TaxiFare> fares = env.fromSource(
                                source,
                                WatermarkStrategy.<TaxiFare>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                                .withTimestampAssigner((fare, ts) -> fare.getEventTime()),
                                "Kafka Fares");

                // 事件驱动伪窗口：按司机每分钟汇总小费
                DataStream<Tuple3<Long, Long, Float>> minuteTips = fares
                                .keyBy(TaxiFare::getDriverId)
                                .process(new PseudoWindow(Duration.ofMinutes(1)));

                minuteTips.print();
                env.execute("Event-driven Minute Tips");
        }
}
